package day05

import scala.actors.{Actor, Future}
import scala.collection.mutable.ListBuffer
import scala.io.Source

/**
  * 用Actor并行编程实现WordCount
  */
object ActorWordCount {
  def main(args: Array[String]): Unit = {
    //存放接收到每个文件的结果数据
    val replys: ListBuffer[Future[Any]] = new ListBuffer[Future[Any]]

    //存放有值的Future里的数据
    val res = new ListBuffer[Map[String,Int]]

    val files = Array("c:/wlan.log","c:/atlog.txt")
    for(file<- files){
      val task = new Task()
      task.start()
      //异步发送消息，有返回值
      val reply = task !! SmTask(file)
      replys += reply
    }

    while(replys.size>0){
      //过来每个Future对象，如果None类型的就过滤掉
      val dones = replys.filter(_.isSet)
      for (done <- dones ){
        res += done.apply().asInstanceOf[Map[String,Int]]
        replys -= done
      }
    }
    //得到全局聚合的数据
    println(res.flatten.groupBy(_._1).mapValues(_.foldLeft(0)(_+_._2)))
  }
}

class Task extends Actor{
  override def act(): Unit = {
    while (true){
      receive({
        case SmTask(file) =>{
          val lines = Source.fromFile(file,"UTF-8").getLines().toList
          val res = lines.flatMap(_.split(" ")).map((_,1)).groupBy(_._1).mapValues(_.size)
          println(res)
          sender ! res
        }
      })
    }
  }
}

case class SmTask(file:String)